package com.wdy.async.method.invocation;

import java.util.Optional;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @ClassName: ThreadAsyncExecutor
 * @Description: TODO
 * @Date: 2019/12/30 15:52
 * @Author: wangdy001
 * @Version: 1.0
 */
public class ThreadAsyncExecutor implements AsyncExecutor {

    private AtomicInteger idx = new AtomicInteger(1);

    @Override
    public <T> AsyncResult<T> startProcess(Callable<T> callable) {
       return startProcess(callable, null);
    }

    @Override
    public <T> AsyncResult<T> startProcess(Callable<T> callable, AsyncCallback<T> asyncCallback) {
        CompletableResult<T> result = new CompletableResult<>(asyncCallback);
        new Thread(() -> {
            try {
                result.setValue(callable.call());
            } catch (Exception e) {
                result.setException(e);
            }
        }, "Thread-name-" + idx.incrementAndGet()).start();
        return result;
    }

    @Override
    public <T> T endProcess(AsyncResult<T> asyncResult) throws InterruptedException, ExecutionException {
        while(!asyncResult.isCompleted()) {
            asyncResult.await();
        }
        return asyncResult.getValue();
    }

    private static class CompletableResult<T> implements AsyncResult<T>{
        private final int RUNNING = 1;
        private final int FAILED = 2;
        private final int COMPLETED = 3;

        final Object lock;
        final Optional<AsyncCallback<T>> callback;

        volatile int state = RUNNING;
        T value;
        Exception exception;

        public CompletableResult(AsyncCallback<T> callback) {
            this.lock = new Object();
            this.callback = Optional.ofNullable(callback);
        }

        void setValue(T value){
            this.value = value;
            this.state = COMPLETED;
            this.callback.ifPresent(callback1 -> callback1.onComplete(value, Optional.<Exception>empty()));
            synchronized (lock) {
                lock.notifyAll();
            }
        }

        void setException(Exception exception){
            this.exception = exception;
            this.state = COMPLETED;
            this.callback.ifPresent(callback1 -> callback1.onComplete(null, Optional.of(exception)));
            synchronized (lock) {
                lock.notifyAll();
            }
        }

        @Override
        public boolean isCompleted() {
            return state > RUNNING;
        }

        @Override
        public T getValue() throws ExecutionException {
            if(state == COMPLETED) {
                return value;
            } else if(state == FAILED) {
                throw new ExecutionException(exception);
            }
            throw new IllegalStateException("Execution not completed yet");
        }

        @Override
        public void await() throws InterruptedException {
            synchronized (lock) {
                while(!isCompleted()) {
                    lock.wait();
                }
            }
        }
    }
}
